package com.gallop.akka.support;

import akka.actor.*;
import akka.cluster.sharding.ClusterSharding;
import akka.cluster.sharding.ClusterShardingSettings;
import akka.cluster.sharding.ShardRegion;
import akka.cluster.singleton.ClusterSingletonManager;
import akka.cluster.singleton.ClusterSingletonManagerSettings;
import akka.cluster.singleton.ClusterSingletonProxy;
import akka.cluster.singleton.ClusterSingletonProxySettings;
import com.gallop.akka.ActorContextSupport;
import com.gallop.akka.AkkaProperties;
import com.gallop.akka.SpringAkkaExtension;
import com.gallop.akka.util.ActorContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.util.StringUtils;

import java.util.Arrays;
import java.util.function.BiFunction;
import java.util.stream.Collectors;

/**
 * Akka spring support to fetch Actor.
 * <p>
 * <pre>
 * usage:
 *
 * public class SomeSpringService extends AkkaSpringSupport {
 * </pre>
 *
 * @author pipi
 */
@Slf4j
public class AkkaSpringSupport implements ActorContextSupport {

    @Autowired
    @Lazy
    private ActorSystem actorSystem;

    @Autowired
    private AkkaProperties akkaProperties;

    /**
     * 创建一个actor 归属actorSystem
     *
     * @param actorClass 创建actor对应的class
     * @return reference
     */
    @Override
    public ActorRef actorOf(Class<? extends Actor> actorClass) {
//        return actorSystem.actorOf(SpringAkkaExtension.SPRING_EXTENSION_PROVIDER.get(actorSystem).props(actorClass), actorClass.getCanonicalName());
        return actorOf(ActorParam.builder().actorClass(actorClass).build());
    }

    /**
     * 创建一个actor 归属actorSystem，并附带参数。
     * <p>
     * <p>
     * 自定义参数在创建的actor中可以通过{@link Props#args()}获取，注意前两个参数固定为：
     * <li>{@link org.springframework.context.ApplicationContext}</li>
     * <li>{@link Class}</li>
     * 从第三个开始为自定义参数。也可以通过{@link ActorContextUtil#getCustomArgs(ActorContext)}获得
     *
     * @param actorClass 创建actor 的class
     * @param args       自定义参数，参数列表为null或长度为0时不会被传递
     * @return reference
     */
    @Override
    public ActorRef actorOf(Class<? extends Actor> actorClass, Object... args) {
        return actorOf(ActorParam.builder()
                .actorClass(actorClass)
                .args(Arrays.stream(args).collect(Collectors.toList()))
                .build());
    }

    /**
     * 创建一个actor，父结点为<code>parentContext</code>的拥有者
     *
     * @param parentContext 父actor的context
     * @param actorClass    创建actor 的class
     * @param args          自定义参数，参数列表为null或长度为0时不会被传递
     * @return reference
     */
    @Override
    public ActorRef actorOf(ActorContext parentContext, Class<? extends Actor> actorClass, Object... args) {
        return actorOf(ActorParam.builder().parentContext(parentContext)
                .actorClass(actorClass)
                .args(Arrays.stream(args).collect(Collectors.toList())).build());
//        if (parentContext == null) {
//            return actorOf(actorClass);
//        }
//        return parentContext.actorOf(SpringAkkaExtension.SPRING_EXTENSION_PROVIDER.get(parentContext.system()).props(actorClass, args), actorClass.getCanonicalName());
    }

    @Override
    public ActorRef actorOf(ActorParam actorParam) {
        ActorContext parentContext = actorParam.getParentContext();
        Class<? extends Actor> actorClass = actorParam.getActorClass();
        String actorName = StringUtils.hasLength(actorParam.getActorName()) ? actorParam.getActorName() : actorClass.getCanonicalName();
        Object[] args = actorParam.getArgs() == null ? null : actorParam.getArgs().toArray();

        BiFunction<Props, String, ActorRef> actorOfInvoker = parentContext == null ?
                actorSystem::actorOf : parentContext::actorOf;
        Props props = systemPropsBuilder(parentContext == null ?
                actorSystem : parentContext.system(), actorClass, args);
        if (actorParam.getPropDecorator() != null) {
            props = actorParam.getPropDecorator().apply(props);
        }
        return actorOfInvoker.apply(props, actorName);
//        if (parentContext == null) {
//            return actorSystem.actorOf(systemPropsBuilder(actorSystem, actorClass, args), actorName);
//        } else {
//            return parentContext.actorOf(systemPropsBuilder(parentContext.system(), actorClass, args), actorName);
//        }
    }

    private Props systemPropsBuilder(ActorSystem actorSystem, Class<? extends Actor> actorClass, Object... args) {
        if (args != null) {
            return SpringAkkaExtension.SPRING_EXTENSION_PROVIDER.get(actorSystem).props(actorClass, args);
        }
        return SpringAkkaExtension.SPRING_EXTENSION_PROVIDER.get(actorSystem).props(actorClass);
    }

    @Override
    public ActorRef singletonOf(Class<? extends Actor> actorClass, Object terminationMessage,
                                Object... args) {
        Props singletonProps = ClusterSingletonManager.props(systemPropsBuilder(actorSystem, actorClass, args),
                terminationMessage, ClusterSingletonManagerSettings.create(actorSystem));
        ActorRef singletonActorRef = actorSystem.actorOf(singletonProps, actorClass.getCanonicalName());
        log.debug("create singleton actor :{}", singletonActorRef.path());
        return actorSystem.actorOf(ClusterSingletonProxy.props(singletonActorRef.path().toStringWithoutAddress(), ClusterSingletonProxySettings.create(actorSystem)),
                singletonActorRef.path().name() + "-singleton-proxy");
    }

    @Override
    public ActorRef singletonOf(Class<? extends Actor> actorClass, Object terminationMessage) {
        Props singletonProps = ClusterSingletonManager.props(systemPropsBuilder(actorSystem, actorClass),
                terminationMessage, ClusterSingletonManagerSettings.create(actorSystem));
        ActorRef singletonActorRef = actorSystem.actorOf(singletonProps, actorClass.getCanonicalName());
        ActorRef proxyActorRef = actorSystem.actorOf(
                ClusterSingletonProxy.props(singletonActorRef.path().toStringWithoutAddress(), ClusterSingletonProxySettings.create(actorSystem)),
                singletonActorRef.path().name() + "-singleton-proxy");
        log.debug("create singleton actor :{}, proxy:{}", singletonActorRef.path(), proxyActorRef.path());
        return proxyActorRef;
    }

    @Override
    public ActorRef clusterShardingOf(Class<? extends Actor> actorClass, ShardRegion.MessageExtractor messageExtractor) {
        ClusterShardingSettings settings = ClusterShardingSettings.create(actorSystem);
//        settings.withRememberEntities(true);
//        settings.withRole(akkaProperties.getClusterRole());
        return ClusterSharding.get(actorSystem).start(
                actorClass.getCanonicalName(), systemPropsBuilder(actorSystem, actorClass),
                settings, messageExtractor);
//        return ClusterSharding.get(actorSystem).startProxy(
//                actorClass.getCanonicalName() + id,
//                Optional.of(akkaProperties.getClusterRole()),
//                messageExtractor);
    }

    @Override
    public ActorSelection actorSelection(String path) {
        return actorSystem.actorSelection(path);
    }

    @Override
    public ActorSelection actorSelection(ActorContext context, String path) {
        return context.actorSelection(path);
    }
}
